package com.csw.flink.cep

import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._

object Demo01Cep {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val ds: DataStream[Event] = env.socketTextStream("master", 8888)
      .map(line => {
        val split: Array[String] = line.split(",")

        Event(split(0), split(1), split(2))
      })

    /**
      * 数据
      * 1,192.168.0.1,a
      * 1,192.168.0.2,b
      * 1,192.168.0.3,c
      * 1,192.168.0.4,d
      *
      * 匹配数据流中 a出现之后出现b
      *
      * 匹配 a出现之后  可选：出现了多个b或者多个c  (2,4)
      *
      * next()，指定严格连续，
      * followedBy()，指定松散连续，
      * followedByAny()，指定不确定的松散连续。
      *
      * notNext()，如果不想后面直接连着一个特定事件
      * notFollowedBy()，如果不想一个特定事件发生在两个事件之间的任何地方。
      */


    //定义匹配模板
    val pattern: Pattern[Event, Event] = Pattern
      .begin[Event]("begin") //匹配第一个事件
      .where(event => event.t.equals("a"))
      .followedBy("next") //松散连续匹配下一个事件
      .where(event => event.t.equals("b"))
      .or(event => event.t.equals("c"))
      .times(2, 4)//2到4个
      .optional //可选模式，可以有也可以没有

    //对数据流进行匹配
    val patternStream: PatternStream[Event] = CEP.pattern(ds.keyBy(_.id), pattern)

    //取出匹配成功的数据
    val resultDS: DataStream[collection.Map[String, Iterable[Event]]] = patternStream.select(m => m)

    //打印结果
    resultDS.print()

    env.execute()
  }
}

case class Event(id: String, ip: String, t: String)
